package com.jltx.rx;

import java.util.ArrayList;
import java.util.List;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2017/08/09.
 */

public class RxJava2Learn2 {

    public  static void main(String[] args){
        RxJava2Learn2 rxjava = new RxJava2Learn2();
//        rxjava.justOperator();
//        rxjava.fromArrayOperator();
//        rxjava.fromIterableOperator();
//        rxjava.mapOperator();
//        rxjava.mapOperator1();
//        rxjava.flatMapOperator();
//        rxjava.filterOperator();
        rxjava.takeOperator();
    }

    public void justOperator() {
        Observable.just("map", "map1", "map2")
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:" + s);
                    }
                });
    }

    public void fromArrayOperator() {
        Observable.fromArray(new String[]{"map", "map1", "map2"})
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:" + s);
                    }
                });
    }

    /**
     * 接收一个 Iterable 容器作为输入，每次发射一个元素
     */
    public void fromIterableOperator() {
        List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(1);
        list.add(20);

        Observable.fromIterable(list)
                .subscribe(num -> System.out.println("accept:" + num));
    }

    /**
     * map的作用就变换 Flowable 然后返回一个指定类型的 Flowable 对象
     */
    public void mapOperator(){
        Observable.just("map","map1","map2")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        return s + "-jltx";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:"+s);
                    }
                });
    }

    /**
     * map 操作符更神奇的地方是，你可以返回任意类型的 Observable，
     * 也就是说你可以使用 map 操作符发射一个新的数据类型的 Observable 对象。
     */
    public void mapOperator1(){
        Observable.just("map","map1","map2")
                .map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String s) throws Exception {
                        return s.hashCode();
                    }
                })
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        return integer.toString()+"-jltx";
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("accept:"+s);
                    }
                });
    }



    /**
     * Flowable.flatMap 可以把一个 Observable 转换成另一个 Observable
     */
    public void flatMapOperator(){
        List<Integer> list = new ArrayList<>();
        list.add(10);
        list.add(1);
        list.add(20);
        Observable.just(list)
                .flatMap(new Function<List<Integer>, ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> apply(List<Integer> integers) throws Exception {
                        return Observable.fromIterable(integers);
                    }
                })
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Exception {
                        System.out.println("map.thread:"+Thread.currentThread());
                        return integer.toString();
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        System.out.println("subscribe.thread:"+Thread.currentThread());
                        System.out.println("accept:"+s);
                    }
                });
    }

    /**
     * filter 是用于过滤数据的，返回false表示拦截此数据。
     */
    public void filterOperator(){
        Observable.fromArray(1,10,6,0,20,-1,8)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Exception {
                        return integer.intValue() > 5;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("accept:"+integer);
                    }
                });
    }

    /**
     * take 用于指定订阅者最多收到多少数据
     */
    public void takeOperator(){
        Observable.fromArray(1,10,6,0,20,-1,8)
                .take(2)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        System.out.println("accept:"+integer);
                    }
                });
    }

}
